package com.relytosoft.mizar.analyze.handle.service.impl;

import com.alibaba.fastjson.JSON;
import com.relytosoft.dubhe.common.elasticsearch.ElasticsearchUtil;
import com.relytosoft.dubhe.common.kafka.producer.KafkaProducer;
import com.relytosoft.dubhe.common.redis.utils.RedisUtil;
import com.relytosoft.mizar.analyze.base.constant.AyCacheConstants;
import com.relytosoft.mizar.analyze.base.entity.algo.AYNonVehicle;
import com.relytosoft.mizar.analyze.base.entity.algo.AYPerson;
import com.relytosoft.mizar.analyze.base.entity.algo.AYVehicle;
import com.relytosoft.mizar.analyze.base.entity.algo.AYVideoSlice;
import com.relytosoft.mizar.analyze.base.entity.pojo.dto.AyControlAlarmDto;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlDataReq;
import com.relytosoft.mizar.analyze.base.entity.pojo.request.btdl.common.BtdlStructureControlReq;
import com.relytosoft.mizar.analyze.base.enums.IndexNameTypeEnum;
import com.relytosoft.mizar.analyze.common.utils.IndexUtil;
import com.relytosoft.mizar.analyze.handle.feign.AlarmFeignService;
import com.relytosoft.mizar.analyze.handle.service.ControlPlanService;
import com.relytosoft.mizar.analyze.handle.service.StructureListenerService;
import com.relytosoft.mizar.analyze.handle.service.UploadImageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static com.relytosoft.mizar.analyze.base.constant.MediaMqConstants.UNIFY_ALARM_TOPIC;
import static com.relytosoft.mizar.analyze.handle.constant.ErrorConstants.*;

/**
 * @author wss
 * @version 1.0
 * @date 2022/5/16 23:02
 */
@Service
public class StructureListenerServiceImpl implements StructureListenerService<BtdlDataReq, BtdlStructureControlReq> {

    private Logger log = LoggerFactory.getLogger(this.getClass());

    static ThreadPoolExecutor threadPoolExecutor;

    static {
        threadPoolExecutor = new ThreadPoolExecutor(
                (int) (Runtime.getRuntime().availableProcessors() * 1.5),
                Runtime.getRuntime().availableProcessors() * 3,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<Runnable>(10000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

    @Autowired
    private UploadImageService uploadImageService;
    @Autowired
    private ElasticsearchUtil elasticsearchUtil;
    @Autowired
    private ControlPlanService controlPlanService;
    @Autowired
    private KafkaProducer kafkaProducer;
    @Autowired
    private AlarmFeignService alarmFeignService;

    @Override
    public Boolean handleRawData(BtdlDataReq baseAlgoReq) {
        try {
            Map<String, String> bigPic = baseAlgoReq.getBigPic();
            Map<String, String> smallPic = baseAlgoReq.getSmallPic();
            CompletableFuture<Void> bigPicFuture = CompletableFuture.runAsync(() -> rawDataUrlReplace(bigPic));
            CompletableFuture<Void> smallPicFuture = CompletableFuture.runAsync(() -> rawDataUrlReplace(smallPic));
            CompletableFuture.allOf(bigPicFuture, smallPicFuture).get();
            return true;
        } catch (Exception e) {
            log.error("{},--", STRUTE_HANDLE_RAW_DATA_ERROR, e);
            return false;
        }
    }

    public void rawDataUrlReplace(Map<String, String> pic) {
        for (String bigPicKey : pic.keySet()) {
            String ImageBaseStr = pic.get(bigPicKey);
            String imageUrl = uploadImageService.base64Upload(ImageBaseStr);
            pic.put(bigPicKey, imageUrl);
        }
    }

    @Override
    public void handleIOData(BtdlDataReq baseAlgoReq) {
        try {
            List<AYVideoSlice> videoSlice = baseAlgoReq.getVideoSliceListObject().getVideoSlice();
            Map<String, String> bigPic = baseAlgoReq.getBigPic();
            Map<String, String> smallPic = baseAlgoReq.getSmallPic();
            videoSlice.stream().forEach(v -> {
                List<AYVehicle> ayVehicles = v.getMotorVehicleList().getMotorVehicleObject();
                List<AYNonVehicle> nonVehicles = v.getNonMotorVehicleList().getNonMotorVehicleObject();
                List<AYPerson> ayPeoples = v.getPersonList().getPersonObject();
                String deviceId = v.getDeviceID();
                threadPoolExecutor.execute(() -> saveVehicleToDb(ayVehicles, bigPic, smallPic, deviceId));
                threadPoolExecutor.execute(() -> saveNoVehicleToDb(nonVehicles, bigPic, smallPic, deviceId));
                threadPoolExecutor.execute(() -> savePeopleToDb(ayPeoples, bigPic, smallPic, deviceId));
            });
        } catch (Exception e) {
            log.error("{},--", STRUTE_HANDLE_IO_DATA_ERROR, e);
        }

    }

    /**
     * 保存机动车数据到 es redis
     *
     * @param ayVehicles
     * @param bigPic
     * @param smallPic
     * @param deviceId
     */
    void saveVehicleToDb(List<AYVehicle> ayVehicles, Map<String, String> bigPic, Map<String, String> smallPic, String deviceId) {
        if (ayVehicles.isEmpty()) {
            return;
        }
        ayVehicles = ayVehicles.stream().map(a -> {
            a.setImage(bigPic.get(a.getImage()));
            a.setThumb(smallPic.get(a.getThumb()));
            a.setDeviceId(deviceId);
            return a;
        }).collect(Collectors.toList());
        String vehicleIndex = IndexUtil.toMonthIndex(IndexNameTypeEnum.AY_STRUCTURE_MOTOR_VEHICLE.getMessage());
        try {
            elasticsearchUtil.addDocmentByBatch(vehicleIndex, null, ayVehicles);
            saveToRedis(ayVehicles.size(), AyCacheConstants.AY_STRUCTURE_VEHICLE_TOTAL_STATISTICS);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存非机动车数据到es redis
     *
     * @param nonVehicles
     * @param bigPic
     * @param smallPic
     * @param deviceId
     */
    void saveNoVehicleToDb(List<AYNonVehicle> nonVehicles, Map<String, String> bigPic, Map<String, String> smallPic, String deviceId) {
        if (nonVehicles.isEmpty()) {
            return;
        }
        nonVehicles = nonVehicles.stream().map(a -> {
            a.setImage(bigPic.get(a.getImage()));
            a.setThumb(smallPic.get(a.getThumb()));
            a.setDeviceId(deviceId);
            return a;
        }).collect(Collectors.toList());
        String noVehicleIndex = IndexUtil.toMonthIndex(IndexNameTypeEnum.AY_STRUCTURE_NON_MOTOR_VEHICLE.getMessage());
        try {
            elasticsearchUtil.addDocmentByBatch(noVehicleIndex, null, nonVehicles);
            saveToRedis(nonVehicles.size(), AyCacheConstants.AY_STRUCTURE_NO_VEHICLE_TOTAL_STATISTICS);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存行人数据到es redis
     *
     * @param ayPeoples
     * @param bigPic
     * @param smallPic
     * @param deviceId
     */
    void savePeopleToDb(List<AYPerson> ayPeoples, Map<String, String> bigPic, Map<String, String> smallPic, String deviceId) {
        if (ayPeoples.isEmpty()) {
            return;
        }
        ayPeoples = ayPeoples.stream().map(a -> {
            a.setImage(bigPic.get(a.getImage()));
            a.setThumb(smallPic.get(a.getThumb()));
            a.setDeviceId(deviceId);
            return a;
        }).collect(Collectors.toList());
        String peopleIndex = IndexUtil.toMonthIndex(IndexNameTypeEnum.AY_STRUCTURE_PERSON.getMessage());
        try {
            elasticsearchUtil.addDocmentByBatch(peopleIndex, null, ayPeoples);
            saveToRedis(ayPeoples.size(), AyCacheConstants.AY_STRUCTURE_PEOPLE_TOTAL_STATISTICS);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 根据key 添加统计数量
     *
     * @param num
     * @param statisticsName
     */
    void saveToRedis(int num, String statisticsName) {
        Integer newSum;
        Object oldSum = RedisUtil.get(statisticsName);
        newSum = Objects.isNull(oldSum)
                ? num
                : Integer.parseInt(oldSum.toString()) + num;
        RedisUtil.set(statisticsName, newSum);
    }

    @Override
    public void handleBusinessData(BtdlDataReq baseAlgoReq) {
        baseAlgoReq.getVideoSliceListObject().getVideoSlice().stream()
                .forEach(ayVideoSlice -> {
                    try {
                        if (controlPlanService.controlStructureAlarm(ayVideoSlice)) {
                            AyControlAlarmDto ayControlAlarmDto = BtdlDataReq.buildControlAlarm(baseAlgoReq);
                            String jsonString = JSON.toJSONString(ayControlAlarmDto);
                            CompletableFuture<Void> alarmSave = CompletableFuture.runAsync(() -> kafkaProducer.send(UNIFY_ALARM_TOPIC, jsonString, true), threadPoolExecutor);
                            CompletableFuture<Void> alarmReport = CompletableFuture.runAsync(() -> alarmFeignService.alarmReport(jsonString), threadPoolExecutor);
                            CompletableFuture.allOf(alarmSave, alarmReport).join();
                        }
                    } catch (Exception e) {
                        log.error("{},--", STRUTE_HANDLE_BUSINESS_DATA_ERROR, e);
                    }
                });
    }

    @Override
    public boolean handleControlRawData(BtdlStructureControlReq algoControlReq) {
        return false;
    }

    @Override
    public void handleControlIOData(BtdlStructureControlReq algoControlReq) {

    }

    @Override
    public void handleControlBusinessData(BtdlStructureControlReq algoControlReq) {

    }

}
